package com.lizhiyu.flink.demo3_sink.mysql;

import com.lizhiyu.flink.model.VideoOrder;
import com.lizhiyu.flink.demo2_source.paramlleism.VideoCustomSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/***
 * 自定义sink 向mysql中进行存储
 */
public class MysqlCustomSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        //流处理、或批处理、这里选择自动(会自己找到适合的方式)
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);
        DataStream<VideoOrder> ds1 =  env.addSource(new VideoCustomSource());
        ds1.print();
        DataStreamSink<VideoOrder> ds2 = ds1.addSink(new MysqlCustomSink()).setParallelism(2);
        //DataStream需要调用execute,可以取个名称
        env.execute("source job");
    }
}
